package com.hy.${packageCode}.shared.distribute;

import com.hy.${packageCode}.shared.distribute.cache.DistCache;
import com.hy.${packageCode}.shared.distribute.cache.DistributionCache;
import com.hy.${packageCode}.shared.distribute.cache.DistributionLocalCache;
import com.hy.${packageCode}.shared.distribute.cache.StandaloneCache;
import com.hy.${packageCode}.shared.distribute.counter.DistCounter;
import com.hy.${packageCode}.shared.distribute.counter.RedissonCounter;
import com.hy.${packageCode}.shared.distribute.election.DistLeaderElection;
import com.hy.${packageCode}.shared.distribute.lock.DelegatingLock;
import com.hy.${packageCode}.shared.distribute.lock.DelegatingReadWriteLock;
import com.hy.${packageCode}.shared.distribute.lock.DistLock;
import com.hy.${packageCode}.shared.distribute.lock.DistReadWriteLock;
import com.hy.${packageCode}.shared.distribute.map.DistMap;
import com.hy.${packageCode}.shared.distribute.map.DistributionMap;
import com.hy.${packageCode}.shared.distribute.rate.DistRateLimiter;
import com.hy.${packageCode}.shared.distribute.rate.DistributionRateLimiter;
import com.hy.${packageCode}.shared.distribute.cache.DistCache;
import com.hy.${packageCode}.shared.distribute.cache.DistributionCache;
import com.hy.${packageCode}.shared.distribute.cache.DistributionLocalCache;
import com.hy.${packageCode}.shared.distribute.cache.StandaloneCache;
import com.hy.${packageCode}.shared.distribute.counter.DistCounter;
import com.hy.${packageCode}.shared.distribute.counter.RedissonCounter;
import com.hy.${packageCode}.shared.distribute.election.DistLeaderElection;
import com.hy.${packageCode}.shared.distribute.election.RedissonLeaderElection;
import com.hy.${packageCode}.shared.distribute.lock.DelegatingLock;
import com.hy.${packageCode}.shared.distribute.lock.DelegatingReadWriteLock;
import com.hy.${packageCode}.shared.distribute.lock.DistLock;
import com.hy.${packageCode}.shared.distribute.lock.DistReadWriteLock;
import com.hy.${packageCode}.shared.distribute.map.DistMap;
import com.hy.${packageCode}.shared.distribute.map.DistributionMap;
import com.hy.${packageCode}.shared.distribute.rate.DistRateLimiter;
import com.hy.${packageCode}.shared.distribute.rate.DistributionRateLimiter;
import com.hy.${packageCode}.shared.distribute.cache.DistCache;
import com.hy.${packageCode}.shared.distribute.cache.DistributionCache;
import com.hy.${packageCode}.shared.distribute.cache.DistributionLocalCache;
import com.hy.${packageCode}.shared.distribute.cache.StandaloneCache;
import com.hy.${packageCode}.shared.distribute.counter.DistCounter;
import com.hy.${packageCode}.shared.distribute.counter.RedissonCounter;
import com.hy.${packageCode}.shared.distribute.election.DistLeaderElection;
import com.hy.${packageCode}.shared.distribute.lock.DelegatingLock;
import com.hy.${packageCode}.shared.distribute.lock.DelegatingReadWriteLock;
import com.hy.${packageCode}.shared.distribute.lock.DistLock;
import com.hy.${packageCode}.shared.distribute.lock.DistReadWriteLock;
import com.hy.${packageCode}.shared.distribute.map.DistMap;
import com.hy.${packageCode}.shared.distribute.map.DistributionMap;
import com.hy.${packageCode}.shared.distribute.rate.DistRateLimiter;
import com.hy.${packageCode}.shared.distribute.rate.DistributionRateLimiter;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.LocalCachedMapOptions;
import org.redisson.api.RLocalCachedMap;
import org.redisson.api.RMap;
import org.redisson.api.RedissonClient;
import org.redisson.client.codec.Codec;

import java.time.Duration;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 基于 redisson(redis) 实现的分布式对象工厂
 */
@Slf4j
public class DistributionObjectFactory implements DistObjectFactory {

    private static final Duration CLEAN_SCHEDULE_PERIOD = Duration.ofHours(2);
    private final ScheduledExecutorService cleanScheduler = Executors.newSingleThreadScheduledExecutor();
    private final Map<String, DistributionLocalCache> distLocalCaches = new ConcurrentHashMap<>();

    private final Map<String, RedissonLeaderElection> leaderElections = new ConcurrentHashMap<>();
    private final RedissonClient redissonClient;
    private final Codec codec;

    public DistributionObjectFactory(RedissonClient redissonClient) {
        this(redissonClient, null);
    }

    public DistributionObjectFactory(RedissonClient redissonClient, Codec codec) {
        this.redissonClient = redissonClient;
        this.codec = codec == null ? redissonClient.getConfig().getCodec() : codec;
        cleanScheduler.scheduleWithFixedDelay(this::runCacheExpiredClean,
                CLEAN_SCHEDULE_PERIOD.toMinutes(), CLEAN_SCHEDULE_PERIOD.toMinutes(), TimeUnit.MINUTES);
    }

    protected void runCacheExpiredClean() {
        long cleanAt = System.currentTimeMillis();
        distLocalCaches.values().forEach(DistributionLocalCache::cleanExpired);
        if (System.currentTimeMillis() > cleanAt + CLEAN_SCHEDULE_PERIOD.toMillis()) {
            log.warn("The scheduler run cache expired clean time reach at {}", CLEAN_SCHEDULE_PERIOD);
        }
    }

    @Override
    public String id() {
        return redissonClient.getId();
    }

    @Override
    @PreDestroy
    public void shutdown() {
        cleanScheduler.shutdownNow();
        leaderElections.values()
                .forEach(RedissonLeaderElection::shutdown);
    }


    @Override
    public <K, V> DistMap<K, V> getMap(String name, boolean withLocalCache) {
        RMap<K, V> delegatingMap;
        if (withLocalCache) {
            delegatingMap = redissonClient.getLocalCachedMap(name, codec, LocalCachedMapOptions.<K, V>defaults()
                    .evictionPolicy(LocalCachedMapOptions.EvictionPolicy.SOFT) //本地缓存 使用  SOFT 淘汰策略
                    .cacheProvider(LocalCachedMapOptions.CacheProvider.CAFFEINE)); //本地缓存 使用 CAFFEINE 实现
        } else {
            delegatingMap = redissonClient.getMap(name, codec);
        }
        return new DistributionMap<>(delegatingMap);
    }

    @Override
    public DistLock getLock(String name) {
        return new DelegatingLock(redissonClient.getLock(name));
    }

    @Override
    public DistReadWriteLock getReadWriteLock(String name) {
        return new DelegatingReadWriteLock(redissonClient.getReadWriteLock(name));
    }

    @Override
    public DistCounter getCounter(String name) {
        return new RedissonCounter(redissonClient.getAtomicLong(name));
    }

    @Override
    public <K, V> DistCache<K, V> getCache(String name, boolean withLocalCache) {
        if (!withLocalCache) {
            return new DistributionCache<>(redissonClient.getMapCache(name, codec));
        }
        return distLocalCaches.computeIfAbsent(name, key -> {
            RLocalCachedMap<K, V> delegatingMap = redissonClient.getLocalCachedMap(name, codec, LocalCachedMapOptions.<K, V>defaults()
                    .evictionPolicy(LocalCachedMapOptions.EvictionPolicy.SOFT) // 本地缓存 使用  SOFT 淘汰策略
                    .cacheProvider(LocalCachedMapOptions.CacheProvider.CAFFEINE)); //本地缓存 使用 CAFFEINE 实现
            RLocalCachedMap<K, StandaloneCache.ExpireItem> expireItems = redissonClient.
                    getLocalCachedMap(name + ".expires", codec, LocalCachedMapOptions.<K, StandaloneCache.ExpireItem>defaults()
                            .evictionPolicy(LocalCachedMapOptions.EvictionPolicy.SOFT) // 本地缓存 使用  SOFT 淘汰策略
                            .cacheProvider(LocalCachedMapOptions.CacheProvider.CAFFEINE)); //本地缓存 使用 CAFFEINE 实现
            return new DistributionLocalCache<>(delegatingMap, expireItems, getReadWriteLock(name + ".ops_lock"));
        });
    }

    @Override
    public DistLeaderElection getLeaderElection(String name) {
        return leaderElections.computeIfAbsent(name,
                key -> new RedissonLeaderElection(redissonClient.getLock(name)));
    }

    @Override
    public DistRateLimiter getRateLimiter(String name) {
        return DistributionRateLimiter.of(redissonClient.getRateLimiter(name));
    }
}
